#include "NetMgr.h"
#include <forward_list>


NetMgr NetMgr::m_oInstance;



WorkerThreadN::WorkerThreadN()
{
}
WorkerThreadN::~WorkerThreadN()
{
	StopRun();
}

void WorkerThreadN::UpdateChannel(ChannelPtr channel)
{
#if 0
	Event tEvent;
	{
		std::unique_lock<std::mutex> lock(_mutex);
		_queue.push_back([this, &channel, &tEvent]() {
			

			SOCKET socket = channel->GetSocket();

			std::unordered_map<SOCKET, ChannelPtr>::iterator iter = channels_.find(socket);
			if (iter != channels_.end()) {
				if (channel->IsNoneEvent()) {

					channels_.erase(socket);
				}
				else {
					// 更新读写事件 暂时不考虑
					//iter->second->
				}
			}
			else {
				if (!channel->IsNoneEvent()) {
					channels_.emplace(socket, channel);

					
				}
				printf("channel :%d \n", channel->GetSocket());
			}


			tEvent.Set(0);
			});
	}
	char event = 1;
	m_oPipe.Write(&event, 1);
	printf("UpdateChannel 0 \n");
	tEvent.Wait();
	printf("UpdateChannel over  \n");
#endif

	InvokTask([this, &channel]()
		{

			SOCKET socket = channel->GetSocket();

			std::unordered_map<SOCKET, ChannelPtr>::iterator iter = channels_.find(socket);
			if (iter != channels_.end()) {
				if (channel->IsNoneEvent()) {

					channels_.erase(socket);
				}
				else {
					// 更新读写事件 暂时不考虑
					//iter->second->
				}
			}
			else {
				if (!channel->IsNoneEvent()) {
					channels_.emplace(socket, channel);


				}
				printf("channel :%d \n", channel->GetSocket());
			}
		}
	);
}

void WorkerThreadN::RemoveChannel(ChannelPtr& channel)
{
#if 0
	Event tEvent;
	{
		std::unique_lock<std::mutex> lock(_mutex);
		_queue.push_back([this, &channel, &tEvent]() {

			SOCKET fd = channel->GetSocket();

			if (channels_.find(fd) != channels_.end()) {
				channels_.erase(fd);
			}
			tEvent.Set(0);

			});
	}
	char event = 1;
	m_oPipe.Write(&event, 1);
	printf("RemoveChannel 0 \n");
	tEvent.Wait();
	printf("RemoveChannel over \n");
#endif

	InvokTask([&channel, this]()
		{
			SOCKET fd = channel->GetSocket();

			if (channels_.find(fd) != channels_.end()) {
				channels_.erase(fd);
			}
		}
	);
}

void WorkerThreadN::PostTask(Task task)
{
	{
		std::unique_lock<std::mutex> lock(_mutex);
		_queue.push_back([this, &task]() {

			task();

			});
	}
	char event = 1;
	m_oPipe.Write(&event, 1);
}


void WorkerThreadN::InvokTask(Task task)
{
	Event tEvent;
	{
		std::unique_lock<std::mutex> lock(_mutex);
		_queue.push_back([task, &tEvent]() {
			task();
			tEvent.Set(0);
			});
	}
	char event = 1;
	m_oPipe.Write(&event, 1);
	printf("Wait before \n");
	tEvent.Wait();
}
int64_t WorkerThreadN::getCurrentTicket()
{
	//std::chrono::steady_clock::time_point t1 = std::chrono::steady_clock::now();
	//std::chrono::steady_clock::time_point t11;
	//int64_t duration = std::chrono::duration_cast<std::chrono::microseconds>(t1 - t11).count();
	//int64_t duration = std::chrono::duration_cast<std::chrono::milliseconds>(t1 - t11).count();
	return  std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
}
int32_t WorkerThreadN::StartRun()
{

	bool ret = m_oPipe.Create();
	if (false == ret)
	{
		return -2;
	}

	m_loopThread = new std::thread(&WorkerThreadN::RunLoop, this);
	std::unique_lock<std::mutex> lock(_mutex);
	_condition.wait(lock, [this]() { return m_bStart == true; });

	if (NULL == m_loopThread)
	{
		return -1;
	}
	
	return 0;

}
void WorkerThreadN::StopRun()
{
	m_bExitThread = true;
	{
		std::unique_lock<std::mutex> lock(_mutex);
		_queue.push_back([]() {int a; });
	}
	_condition.notify_one();

	if (m_loopThread)
	{
		m_loopThread->join();
		delete m_loopThread;
		m_loopThread = 0;
	}

	m_oPipe.Close();
}



void WorkerThreadN::RunLoop()
{
	_loopThreadId = std::this_thread::get_id();
	m_bStart = true;
	_condition.notify_one();

	fd_set fd_read;
	fd_set fd_write;
	fd_set fd_exp;


	while (!m_bExitThread)
	{
		int max_fd = 0;

		FD_ZERO(&fd_read);
		FD_ZERO(&fd_write);
		FD_ZERO(&fd_exp);


		FD_SET(m_oPipe.Read(), &fd_read);


		max_fd = m_oPipe.Read();

		for (auto iter : channels_) {
			int events = iter.second->GetEvents();
			SOCKET fd = iter.second->GetSocket();

			if ((events & EVENT_IN)) {
				FD_SET(fd, &fd_read);
			}

			if ((events & EVENT_OUT)) {
				FD_SET(fd, &fd_write);
			}
			if (fd > max_fd)
			{
				max_fd = fd;
			}

		}

		int ret = select(max_fd + 1, &fd_read, &fd_write, NULL, NULL);

		if (ret > 0)
		{

			std::forward_list<std::pair<ChannelPtr, int>> event_list;
			 {

				for (auto iter : channels_) {
					int events = 0;
					SOCKET socket = iter.second->GetSocket();

					if (FD_ISSET(socket, &fd_read)) {
						events |= EVENT_IN;
					}

					if (FD_ISSET(socket, &fd_write)) {
						events |= EVENT_OUT;
					}

					if (FD_ISSET(socket, &fd_exp)) {
						events |= (EVENT_HUP); // close
					}

					if (events != 0) {
						event_list.emplace_front(iter.second, events);
					}
				}
			}

			for (auto& iter : event_list) {
				iter.first->HandleEvent(iter.second);
			}
			
			if (FD_ISSET(m_oPipe.Read(), &fd_read))
			{
				char buf[1];
				m_oPipe.Read(buf, sizeof(buf));
				// 处理增删任务
				std::list<Task> t_queue;
				{
					std::unique_lock<std::mutex> lock(_mutex);
					if (!_queue.empty())
					{
						t_queue.swap(_queue);

					}
				}

				// 处理任务
				if (!t_queue.empty())
				{
					for (std::list<Task>::iterator it = t_queue.begin(); it != t_queue.end(); it++)
					{
						Task task = *it;
						task();
					}
					t_queue.clear();
				}
				printf("处理管道任务\n");
			}


		}
		//for (auto& pr : m_event_map) {

		//	shared_ptr<timer> mT = pr.second;
		//	if (FD_ISSET(mT->GetSocketfd(), (fd_set*)&read_set))
		//	{

		//		// 读取数据放在回调中

		//		mT->get_user_callback_f()();
		//	}
		//}

#if 0
		std::list<Task> t_queue;
		{
			//std::unique_lock<std::mutex> lock(_mutex);
			////_condition.wait_for(lock, std::chrono::milliseconds(delay), [this]() { return !_queue.empty(); });
			//_condition.wait_for(lock, std::chrono::microseconds(delay), [this]() { return !_queue.empty(); });
			//if (!_queue.empty())
			//{
			//	t_queue.swap(_queue);

			//}
		}
		// 处理任务
		if (!t_queue.empty())
		{
			for (std::list<Task>::iterator it = t_queue.begin(); it != t_queue.end(); it++)
			{
				Task task = *it;
				task();
			}
			t_queue.clear();
		}

#endif
	}

}

EventLoopN::EventLoopN(uint32_t num_threads)
{

}

EventLoopN::~EventLoopN()
{
}

void EventLoopN::Loop()
{
	std::lock_guard<std::mutex> locker(mutex_);

	if (!task_schedulers_.empty()) {
		return;
	}

	for (uint32_t n = 0; n < num_threads_; n++)
	{
		std::shared_ptr<WorkerThreadN> task_scheduler_ptr(new WorkerThreadN());
		task_scheduler_ptr->StartRun();
		task_schedulers_.push_back(task_scheduler_ptr);

	}
}

void EventLoopN::Quit()
{
	std::lock_guard<std::mutex> locker(mutex_);

	for (auto iter : task_schedulers_) {
		iter->StopRun();
	}
	task_schedulers_.clear();
}

std::shared_ptr<WorkerThreadN> EventLoopN::GetTaskScheduler()
{
	std::lock_guard<std::mutex> locker(mutex_);
	if (task_schedulers_.size() == 1) {
		return task_schedulers_.at(0);
	}
	else {
		auto task_scheduler = task_schedulers_.at(index_);
		index_++;
		if (index_ >= task_schedulers_.size()) {
			index_ = 1;
		}
		return task_scheduler;
	}

	return nullptr;
}

void EventLoopN::UpdateChannel(ChannelPtr channel)
{
	std::lock_guard<std::mutex> locker(mutex_);
	if (task_schedulers_.size() > 0) {
		task_schedulers_[0]->UpdateChannel(channel);
	}
}

void EventLoopN::RemoveChannel(ChannelPtr& channel)
{
	std::lock_guard<std::mutex> locker(mutex_);
	if (task_schedulers_.size() > 0) {
		task_schedulers_[0]->RemoveChannel(channel);
	}
}

void EventLoopN::PostTask(Task task)
{

}

NetMgr::NetMgr()
{
}

NetMgr* NetMgr::GetInstance()
{
	return &(NetMgr::m_oInstance);
	
}

int32_t NetMgr::Init(uint32_t num_threads)
{
	m_oEventLoop.Loop();
	return int32_t(0);
}

NetMgr::~NetMgr()
{
}
